﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.DataStruct;
using UtilZ.Dotnet.Ex.Log;

namespace UtilZ.Dotnet.Ex.SequenceDB
{
    /// <summary>
    /// 数据回放
    /// </summary>
    public class SequenceDataReplay : IDisposable
    {
        private readonly string _drid;
        private readonly SingleTypeObjectPool<ISequenceData> _bufferDataPool;
        private readonly SequenceDatabase _database;
        private readonly ISequenceDataFilter _filter;
        private readonly Action<ISequenceData> _replayDataOuputCallback;
        private readonly Action<string, long, object> _replayEndCallback;
        private readonly ThreadEx _replayThread;

        private float _replaySpeed = 1.0f;
        /// <summary>
        /// 回放速度，正常时间的倍数。小于0减速，大于0加速
        /// </summary>
        public float ReplaySpeed
        {
            get { return this._replaySpeed; }
            set { this._replaySpeed = value; }
        }


        /// <summary>
        /// 构造函数初始化
        /// </summary>
        /// <param name="drid">回放标志</param>
        /// <param name="pool">数据对象池</param>
        /// <param name="database">数据存储对象</param>
        /// <param name="filter">数据过滤器，为null表示只通过索引过滤</param>
        /// <param name="replayDataOuputCallback">回放数据输出回调</param>
        /// <param name="replayEndCallback">回放完成通知回调</param>
        /// <exception cref="ArgumentNullException">有参数为null时抛出的异常</exception>
        public SequenceDataReplay(string drid, SingleTypeObjectPool<ISequenceData> pool, SequenceDatabase database,
            ISequenceDataFilter filter, Action<ISequenceData> replayDataOuputCallback, Action<string, long, object> replayEndCallback)
        {
            if (pool == null)
            {
                throw new ArgumentNullException(nameof(pool));
            }

            if (database == null)
            {
                throw new ArgumentNullException(nameof(database));
            }

            if (replayDataOuputCallback == null)
            {
                throw new ArgumentNullException(nameof(replayDataOuputCallback));
            }


            this._drid = drid;
            this._bufferDataPool = pool;
            this._database = database;
            this._filter = filter;
            this._replayDataOuputCallback = replayDataOuputCallback;
            this._replayEndCallback = replayEndCallback;
            this._replayThread = new ThreadEx(this.ReplayThreadMethod, $"数据回放服务_{drid}");
        }

        private void ReplayThreadMethod(ThreadPara threadPara)
        {
            try
            {
                using (var replayPara = (ReplayPara)threadPara.Para)
                {
                    int replayDataIndex = 0;
                    int readCount = 20000;
                    Queue<ISequenceData> queue = new Queue<ISequenceData>(readCount);
                    long total = 0L;

                    while (!threadPara.Token.IsCancellationRequested)
                    {
                        try
                        {
                            replayDataIndex = this.GetData(threadPara.Token, replayPara.ReplayDatas, replayDataIndex, queue, readCount, replayPara.BeginIndex, replayPara.EndIndex);

                            if (queue.Count == 0)
                            {
                                Loger.Info("回放数据结束.");
                                break;
                            }

                            total += queue.Count;
                            this.PrimitiveReplay(threadPara, queue);
                            break;
                        }
                        catch (Exception ex)
                        {
                            Loger.Error(ex);
                        }
                    }

                    if (!threadPara.Token.IsCancellationRequested && this._replayEndCallback != null)
                    {
                        this._replayEndCallback(this._drid, total, replayPara.Tag);
                    }
                }
            }
            catch (Exception ex)
            {
                Loger.Error(ex);
            }
        }

        private int GetData(CancellationToken token, List<ReplayItem> replayDatas, int replayDataIndex, Queue<ISequenceData> queue, int readCount, long dataBeginIndex, long dataEndIndex)
        {
            ISequenceData bufferDataBlock = null;
            for (int index = replayDataIndex; index < replayDatas.Count; index++)
            {
                ReplayItem replayData = replayDatas[index];

                while (!token.IsCancellationRequested)
                {
                    if (bufferDataBlock == null)
                    {
                        if (!this._bufferDataPool.TryTake(out bufferDataBlock))
                        {

                        }
                    }

                    if (this._filter != null)
                    {
                        bool readEnd = true;
                        while (!token.IsCancellationRequested)
                        {
                            if (replayData.TryReadData(bufferDataBlock, dataBeginIndex, dataEndIndex))
                            {
                                if (this._filter.Filter(bufferDataBlock))
                                {
                                    queue.Enqueue(bufferDataBlock);
                                    bufferDataBlock = null;
                                    readEnd = false;
                                    break;
                                }
                            }
                            else
                            {
                                //当前数据读取结束,换个文件
                                break;
                            }
                        }

                        if (readEnd)
                        {
                            break;
                        }
                        else
                        {
                            if (queue.Count >= readCount)
                            {
                                break;
                            }
                        }
                    }
                    else
                    {
                        if (replayData.TryReadData(bufferDataBlock, dataBeginIndex, dataEndIndex))
                        {
                            queue.Enqueue(bufferDataBlock);
                            bufferDataBlock = null;

                            if (queue.Count >= readCount)
                            {
                                break;
                            }
                        }
                        else
                        {
                            //当前数据读取结束,换个文件
                            break;
                        }
                    }
                }
            }

            if (bufferDataBlock != null)
            {
                this._bufferDataPool.Add(bufferDataBlock);
            }

            return replayDatas.Count;
        }

        private void PrimitiveReplay(ThreadPara threadPara, Queue<ISequenceData> queue)
        {
            ISequenceData bufferData;
            long? lastIndex = null;
            long currentIndex;
            long lastOutputIndex = 0L;
            int interval;

            while (!threadPara.Token.IsCancellationRequested && queue.Count > 0)
            {
                if (!queue.TryDequeue(out bufferData))
                {
                    continue;
                }

                currentIndex = bufferData.DataIndex;
                if (lastIndex == null)
                {
                    this._replayDataOuputCallback(bufferData);
                    this._bufferDataPool.Add(bufferData);
                }
                else
                {
                    interval = (int)(((currentIndex - lastIndex.Value)) - (TimeEx.GetTimestamp() - lastOutputIndex));
                    if (interval > 0)
                    {
                        if (this._replaySpeed > 1.0001f)
                        {
                            interval = (int)(interval / this._replaySpeed);
                        }

                        if (interval > 0)
                        {
                            if (threadPara.WaitOne(interval))
                            {
                                break;
                            }
                        }
                    }

                    this._replayDataOuputCallback(bufferData);
                    this._bufferDataPool.Add(bufferData);
                }

                lastIndex = currentIndex;
                lastOutputIndex = TimeEx.GetTimestamp();
            }

            if (threadPara.Token.IsCancellationRequested && queue.Count > 0)
            {
                this._bufferDataPool.Add(queue);
                queue.Clear();
            }
        }


        /// <summary>
        /// 设置回放索引范围
        /// </summary>
        /// <param name="beginIndex">起始索引</param>
        /// <param name="endIndex">截止索引</param>
        /// <param name="tag">附加数据</param>
        /// <returns>在该索引范围内回放的数据文件数量</returns>
        public int SetIndexRange(long beginIndex, long endIndex, object tag)
        {
            this._replayThread.Stop();

            List<ReplayItem> replayDatas = this._database.GetReplayData(beginIndex, endIndex);
            if (replayDatas == null ||
                replayDatas.Count == 0)
            {
                return 0;
            }
            else
            {
                this._replayThread.Start(new ReplayPara(replayDatas, beginIndex, endIndex, tag));
                return replayDatas.Count;
            }
        }


        /// <summary>
        /// 释放资源
        /// </summary>
        public void Dispose()
        {
            this._replayThread.Dispose();
        }

    }

    internal class ReplayPara : IDisposable
    {
        public readonly List<ReplayItem> ReplayDatas;
        public readonly long BeginIndex;
        public readonly long EndIndex;
        public readonly object Tag;

        public ReplayPara(List<ReplayItem> replayDatas, long beginIndex, long endIndex, object tag)
        {
            this.ReplayDatas = replayDatas;
            this.BeginIndex = beginIndex;
            this.EndIndex = endIndex;
            this.Tag = tag;
        }

        public void Dispose()
        {
            foreach (var item in this.ReplayDatas)
            {
                item.Dispose();
            }

            this.ReplayDatas.Clear();
        }
    }

    unsafe internal class ReplayItem : IDisposable
    {
        private readonly FileStream _stream;
        private readonly DataItemHead* _dataItemHeadPtr;
        private readonly int _count;


        public ReplayItem(FileStream stream, DataItemHead* dataItemHeadPtr, int count)
        {
            this._stream = stream;
            this._dataItemHeadPtr = dataItemHeadPtr;
            this._count = count;
        }


        private int _index = 0;
        public bool TryReadData(ISequenceData bufferData, long dataBeginIndex, long dataEndIndex)
        {
            if (this._index >= this._count)
            {
                return false;
            }

            DataItemHead* dataItemHeadPtr = this._dataItemHeadPtr + this._index;
            if (dataItemHeadPtr->Index < dataBeginIndex || dataItemHeadPtr->Index > dataEndIndex)
            {
                return false;
            }

            this._stream.Position = dataItemHeadPtr->DataPosition;
            if (bufferData.TryRead(this._stream, dataItemHeadPtr->DataLength))
            {
                this._index++;
                return true;
            }
            else
            {
                return false;
            }
        }

        public void Dispose()
        {
            this._stream.Dispose();
        }

    }

}
